/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.query.predicate;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
import java.util.logging.Logger;

import edu.ucsd.db.bassarisk.query.predicate.Ranges.Range;

public class PositionPredicates {
  public static class BindingConditionalPredicate extends PositionPredicate {
    private final PositionDeltaRangePredicate predicate;
    private final BitSet requiredBindingIds;

    public BindingConditionalPredicate(final BitSet guard,
                                       final PositionDeltaRangePredicate predicate) {
      this.requiredBindingIds = guard;
      this.predicate = predicate;
    }

    @Override
    public List<PositionPredicate> combine(final PositionPredicate other) {
      if (other instanceof PositionDeltaRangePredicate) {
        PositionDeltaRangePredicate dOther = (PositionDeltaRangePredicate) other;
        PositionPredicate newThis =
            new BindingConditionalPredicate(this.requiredBindingIds,
                                            dOther.combine(this.predicate));
        return list(other, newThis);
      } else if (other instanceof BindingConditionalPredicate) {
        // There are three cases:
        // 1: The predicates apply under the same conditions
        // ___* Predicates may be fully combined
        // 2: One predicate applies under a subset of the other's conditions
        // ___* Looser conditions predicate must be maintained
        // ___* Tighter conditions predicate may have its sub predicate combined
        // ____ with that of the looser
        // 3: Neither set of conditions is a subset of the other
        // ___* No combination is possible
        BindingConditionalPredicate cOther = (BindingConditionalPredicate) other;
        BitSet localConditions = this.getConditionSet();
        localConditions.andNot(cOther.getConditionSet());
        boolean localIsSubset = localConditions.isEmpty();
        BitSet otherConditions = cOther.getConditionSet();
        otherConditions.andNot(this.getConditionSet());
        boolean otherIsSubset = otherConditions.isEmpty();
        if (localIsSubset || otherIsSubset) {
          // This is case 1, 2 or 3
          PositionDeltaRangePredicate newInnerPred =
              this.predicate.combine(cOther.predicate);
          if (localIsSubset && otherIsSubset) {
            // Case 1
            return list(new BindingConditionalPredicate(this.requiredBindingIds,
                                                        newInnerPred));
          } else if (localIsSubset) {
            // Case 2
            return list(this,
                        new BindingConditionalPredicate(cOther.requiredBindingIds,
                                                        newInnerPred));
          } else if (otherIsSubset) {
            // Case 2
            return list(other,
                        new BindingConditionalPredicate(this.requiredBindingIds,
                                                        newInnerPred));
          }
        } else {
          // Case 3
          return list(this, other);
        }
      }
      String msg = "InternalError: Unexpected case.";
      LOG.severe(msg);
      throw new RuntimeException(msg);
    }

    @Override
    protected final int getBindingOne() {
      return this.predicate.getBindingOne();
    }

    @Override
    protected final int getBindingTwo() {
      return this.predicate.getBindingTwo();
    }

    private BitSet getConditionSet() {
      return (BitSet) this.requiredBindingIds.clone();
    }

    @Override
    protected Range getValidDeltaRange() {
      return this.predicate.getValidDeltaRange();
    }

    @Override
    public String toString() {
      return this.predicate.toString() + " when chosen " + this.requiredBindingIds;
    }
  }

  /**
   * A predicate that is satisfied when the delta between two values
   * (represented by two bindings) falls within a particular integer range.
   * 
   * Delta = bindingTwo.boundPosition() - bindingOne.boundPosition() Satisfied
   * if validDeltaRange.contains(Delta)
   */
  public static class PositionDeltaRangePredicate extends PositionPredicate {
    private final int bindingOne;
    private final int bindingTwo;
    private final Range validDeltaRange;

    public PositionDeltaRangePredicate(final int bindingOne, final int bindingTwo,
                                       final Range deltaRange) {
      if (bindingOne < bindingTwo) {
        this.bindingOne = bindingOne;
        this.bindingTwo = bindingTwo;
        this.validDeltaRange = deltaRange;
      } else {
        this.bindingOne = bindingTwo;
        this.bindingTwo = bindingOne;
        this.validDeltaRange = deltaRange.mirror();
      }
    }

    public PositionDeltaRangePredicate
        combine(final PositionDeltaRangePredicate other) {
      assert this.bindingOne == other.getBindingOne();
      assert this.bindingTwo == other.getBindingTwo();
      Range range = Range.intersect(this.validDeltaRange, other.validDeltaRange);
      return new PositionDeltaRangePredicate(this.bindingOne, this.bindingTwo, range);
    }

    @Override
    public List<PositionPredicate> combine(final PositionPredicate other) {
      if (other instanceof PositionDeltaRangePredicate) {
        return list(this.combine((PositionDeltaRangePredicate) other));
      } else if (other instanceof BindingConditionalPredicate) {
        BindingConditionalPredicate cOther = (BindingConditionalPredicate) other;
        PositionPredicate newOther =
            new BindingConditionalPredicate(cOther.requiredBindingIds,
                                            this.combine(cOther.predicate));
        return list(this, newOther);
      }
      String msg = "InternalError: Unexpected case.";
      LOG.severe(msg);
      throw new RuntimeException(msg);
    }

    @Override
    protected final int getBindingOne() {
      return this.bindingOne;
    }

    @Override
    protected final int getBindingTwo() {
      return this.bindingTwo;
    }

    @Override
    protected final Range getValidDeltaRange() {
      return this.validDeltaRange;
    }

    @Override
    public String toString() {
      String format = "$%s ^ $%s in %s";
      return String.format(format, this.bindingOne, this.bindingTwo,
                           this.validDeltaRange.toString());
    }
  }

  public static abstract class PositionPredicate implements
      Comparable<PositionPredicate> {
    public final boolean canCombine(final PositionPredicate other) {
      return this.getBindingOne() == other.getBindingOne()
          && this.getBindingTwo() == other.getBindingTwo();
    }

    public abstract List<PositionPredicate> combine(PositionPredicate other);

    @Override
    public final int compareTo(final PositionPredicate other) {
      int comp = Integer.compare(this.getBindingOne(), other.getBindingOne());
      if (comp == 0) {
        comp = Integer.compare(this.getBindingTwo(), other.getBindingTwo());
      }
      return comp;
    }

    protected abstract int getBindingOne();

    protected abstract int getBindingTwo();

    protected abstract Range getValidDeltaRange();
  }

  static Logger LOG = Logger.getLogger(PositionPredicates.class.getName());

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; Math.abs(bindingTwo.boundValue - bindingOne.boundValue) =
   * delta
   */
  public static PositionDeltaRangePredicate
      createAbsoluteDeltaEquals(final int bindingOne, final int bindingTwo,
                                final int delta) {
    int p = Math.abs(delta);
    int n = -p;
    Range range = Range.union(Range.over(n, n), Range.over(p, p));
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; Math.abs(bindingTwo.boundValue - bindingOne.boundValue) >
   * delta
   */
  public static PositionDeltaRangePredicate
      createAbsoluteDeltaGreater(final int bindingOne, final int bindingTwo,
                                 final int delta) {
    int p = Math.abs(delta);
    int n = -p;
    Range range =
        Range.union(Range.over(Integer.MIN_VALUE, n),
                    Range.over(p, Integer.MAX_VALUE));
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; Math.abs(bindingTwo.boundValue - bindingOne.boundValue) <
   * delta
   */
  public static PositionDeltaRangePredicate
      createAbsoluteDeltaLess(final int bindingOne, final int bindingTwo,
                              final int delta) {
    int p = Math.abs(delta);
    int n = -p;
    Range range = Range.over(n, p);
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  public static BindingConditionalPredicate
      createBindingConditional(final BitSet guard,
                               final PositionDeltaRangePredicate innerPred) {
    return new BindingConditionalPredicate(guard, innerPred);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; bindingTwo.boundValue - bindingOne.boundValue = delta
   */
  public static PositionDeltaRangePredicate createDeltaEquals(final int bindingOne,
                                                              final int bindingTwo,
                                                              final int delta) {
    Range range = Range.over(delta, delta);
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; bindingTwo.boundValue - bindingOne.boundValue > delta
   */
  public static PositionDeltaRangePredicate createDeltaGreater(final int bindingOne,
                                                               final int bindingTwo,
                                                               final int delta) {
    Range range = Range.over(delta, Integer.MAX_VALUE);
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; bindingTwo.boundValue - bindingOne.boundValue < delta
   */
  public static PositionDeltaRangePredicate createDeltaLess(final int bindingOne,
                                                            final int bindingTwo,
                                                            final int delta) {
    Range range = Range.over(Integer.MIN_VALUE, delta);
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; bindingTwo.boundValue - bindingOne.boundValue < 0
   */
  public static PositionDeltaRangePredicate createGreaterThan(final int bindingOne,
                                                              final int bindingTwo) {
    Range range = Range.over(Integer.MIN_VALUE, -1);
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  /**
   * Predicate satisfied when: <br>
   * &nbsp;&nbsp; bindingTwo.boundValue - bindingOne.boundValue > 0
   */
  public static PositionDeltaRangePredicate createLessThan(final int bindingOne,
                                                           final int bindingTwo) {
    Range range = Range.over(1, Integer.MAX_VALUE);
    return new PositionDeltaRangePredicate(bindingOne, bindingTwo, range);
  }

  public static List<PositionPredicate> list(final PositionPredicate... preds) {
    List<PositionPredicate> list = new ArrayList<PositionPredicate>();
    for (PositionPredicate pred : preds) {
      list.add(pred);
    }
    return list;
  }
}
